/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cz.mzk.oai.metadataRepository.harvest.marc21;

import cz.mzk.kramerius.commons.Dom4jUtils;
import cz.mzk.oai.metadataRepository.metadata.DateStamp;
import cz.mzk.oai.metadataRepository.metadata.MetadataFormat;
import cz.mzk.oai.metadataRepository.repository.AlephBase;
import cz.mzk.oai.metadataRepository.repository.Identifier;
import cz.mzk.oai.metadataRepository.repository.Repository;
import cz.mzk.oai.metadataRepository.repository.records.NoSuchRecordException;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.dom4j.Attribute;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.DocumentFactory;
import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.dom4j.Namespace;
import org.dom4j.QName;
import org.dom4j.XPath;
import org.dom4j.io.SAXReader;
import org.marc4j.MarcException;
import org.marc4j.MarcReader;
import org.marc4j.MarcStreamReader;
import org.marc4j.MarcWriter;
import org.marc4j.MarcXmlWriter;
import org.marc4j.marc.ControlField;
import org.marc4j.marc.DataField;
import org.marc4j.marc.Record;

/**
 *
 * @author Martin Rehanek, Martin.Rehanek@gmail.com
 */
public class AlephExportHarvester extends MarcHarvester {

    private static final Logger logger = Logger.getLogger(AlephExportHarvester.class.getName());
    private final int RUNNING_COUNTERS_INTERVAL = 1000;
    private final String CONTROLFIELD_003 = "003";
    private final File dirWithExports;
    private boolean processed = false;
    private int processedCounter = 0;
    private int addedCounter = 0;
    private int deletedCounter = 0;
    private XPath marcRecord = Dom4jUtils.createXPath("/collection/record|/marc21:collection/marc21:record");
    private Namespace xsi = DocumentHelper.createNamespace("xsi", "http://www.w3.org/2001/XMLSchema-instance");
    private String marcSchemaLocation = MetadataFormat.marc21.getNamespaceUri() + " " + MetadataFormat.marc21.getSchemaUrl();

    public AlephExportHarvester(AlephBase base, File dirWithExports, Repository repository, File harvestLog) {
        super(repository, base);
        this.dirWithExports = dirWithExports;
    }

    public void harvest(DateStamp from, DateStamp until) {
        if (processed) {
            throw new IllegalStateException("allready processed, create new instance for new harvest");
        }
        if (from == null) {
            from = DateStamp.MIN;
        }
        if (until == null) {
            until = DateStamp.MAX;
        }
        AlephExports exports = new AlephExports(dirWithExports, alephBase);
        List<AlephExport> relevantExports = exports.getSortedExports(from, until);
        for (AlephExport export : relevantExports) {
            logger.log(Level.INFO, "processing export {0}", export.getFile().getAbsolutePath());
            processExport(export);
        }
        processed = true;
    }

    private void processExport(AlephExport export) {
        InputStream in = null;
        try {
            in = new FileInputStream(export.getFile());
            MarcReader reader = new MarcStreamReader(in, "UTF-8");
            while (reader.hasNext()) {
                processNextRecord(reader);
                printRunningCounters();
            }
            printCounters();
        } catch (IOException ex) {
            logger.log(Level.SEVERE, null, ex);
        } finally {
            try {
                in.close();
            } catch (IOException ex) {
                logger.log(Level.SEVERE, null, ex);
            }
        }


    }

    private void processNextRecord(MarcReader reader) {
        try {
            Record record = reader.next();
            if (isDeletedInAleph(record)) {
                setRecordDeletedInRepository(record);
                System.err.println("setting deleted record");
            } else {
                addOrUpdateRecordInRepository(record);
                System.err.println("adding or updating record");
            }
        } catch (MarcException e) {
            logger.log(Level.SEVERE, null, e);
        } catch (DocumentException ex) {
            logger.log(Level.SEVERE, null, ex);
        } catch (IOException ex) {
            logger.log(Level.SEVERE, null, ex);
        }
        processedCounter++;
    }

    private boolean isDeletedInAleph(Record record) {
        //TODO: poresit s Tomasem, jestli je 003 vzdy pritomen u nesmazanych zaznamu
        //pokud ne, hledat datafield DEL
        //dela to bordel. Napr:
//        SEVERE: null
//        cz.mzk.oai.metadataRepository.repository.records.NoSuchRecordException: id: oai:aleph.mzk.cz:MZK03-001172499, format: marc21
//        at cz.mzk.oai.metadataRepository.repository.Repository.setRecordDeleted(Repository.java:144)
//        at cz.mzk.oai.metadataRepository.harvest.marc21.MarcHarvester.setRecordDeletedInRepository(MarcHarvester.java:36)
//        at cz.mzk.oai.metadataRepository.harvest.marc21.AlephExportHarvester.setRecordDeletedInRepository(AlephExportHarvester.java:154)
//        at cz.mzk.oai.metadataRepository.harvest.marc21.AlephExportHarvester.processNextRecord(AlephExportHarvester.java:100)
//        at cz.mzk.oai.metadataRepository.harvest.marc21.AlephExportHarvester.processExport(AlephExportHarvester.java:86)
//        at cz.mzk.oai.metadataRepository.harvest.marc21.AlephExportHarvester.harvest(AlephExportHarvester.java:75)
//        at cz.mzk.oai.metadataRepository.cli.alephHarvest.ExportHarvester.harvest(ExportHarvester.java:49)
//        at cz.mzk.oai.metadataRepository.cli.alephHarvest.ExportHarvester.harvestFrom(ExportHarvester.java:39)
//        at cz.mzk.oai.metadataRepository.cli.HarvestMarcNewAction.run(HarvestMarcNewAction.java:31)
//        at cz.mzk.oai.metadataRepository.Main.main(Main.java:93)
        List<ControlField> controlFields = record.getControlFields();
        for (ControlField controlField : controlFields) {
            String tag = controlField.getTag();
            if (CONTROLFIELD_003.equals(tag)) {
                return false;
            }
        }
        return true;
    }

    private void printRunningCounters() {
        if (processedCounter % RUNNING_COUNTERS_INTERVAL == 0) {
            printCounters();
        }
    }

    private void printCounters() {
        logger.log(Level.INFO, "counter: {0}, addedCounter: {1}, deletedCounter: {2}", new Object[]{processedCounter, addedCounter, deletedCounter});
    }

    private void addOrUpdateRecordInRepository(Record record) throws DocumentException, IOException {
        boolean removedDatafields = removeDatafieldsWithEmptySubfields(record);
        Identifier id = findId(record);
        if (removedDatafields) {
            logger.log(Level.WARNING, "Record {0} contained datafields with empty subfield content", id.toString());
        }
        Document document = recordToDocument(record);
        Document withoutEnvelope = removeCollectionEnvelopeByMakingDocumentCopy(document);
        addOrUpdateRecordInRepository(id, withoutEnvelope);
        addedCounter++;
    }

    private Identifier findId(Record record) {
        return controlfieldsToId(record.getControlFields(), alephBase);
    }

    private void setRecordDeletedInRepository(Record record) throws IOException {
        Identifier id = findId(record);
        try {
            setRecordDeletedInRepository(id);
            deletedCounter++;
        } catch (NoSuchRecordException ex) {
            Logger.getLogger(AlephExportHarvester.class.getName()).log(Level.SEVERE, id.toString(), ex);
        }
    }

    /**
     * According to the http://www.loc.gov/MARC21/slim
     * http://www.loc.gov/standards/marcxml/schema/MARC21slim.xsd every data
     * field must have at least one subfield. This method removes from record
     * all the datafield with no subfiels so that valid xml in marcxml was
     * created.
     *
     * @param record
     * @return true if something was removed
     */
    private boolean removeDatafieldsWithEmptySubfields(Record record) {
        List<DataField> emptyDataFields = getDatafieldsWithEmptySubfields(record);
        for (DataField dataField : emptyDataFields) {
            record.removeVariableField(dataField);
        }
        return !emptyDataFields.isEmpty();
    }

    List<DataField> getDatafieldsWithEmptySubfields(Record record) {
        List allDatafields = record.getDataFields();
        List<DataField> result = new ArrayList<DataField>(allDatafields.size());
        for (Object obj : allDatafields) {
            DataField dataField = (DataField) obj;
            List subFields = dataField.getSubfields();
            if (subFields.isEmpty()) {
                result.add(dataField);
            }
        }
        return result;
    }

    private Document recordToDocument(Record record) throws DocumentException {
        SAXReader reader = new SAXReader();
        byte[] recordBytes = recordToByteArray(record);
        InputStream in = new ByteArrayInputStream(recordBytes);
        return reader.read(in);
    }

    private byte[] recordToByteArray(Record record) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        MarcWriter writer = new MarcXmlWriter(out, true);
        //write008(record);
        writer.write(record);
        writer.close();
        return out.toByteArray();
    }

    private void write008(Record record) {
        List<ControlField> controlfields = record.getControlFields();
        for (ControlField cf : controlfields) {
            if ("008".equals(cf.getTag())) {
                System.out.println("008: '" + cf.getData() + "'");
            }
        }
    }

    private Identifier controlfieldsToId(List<ControlField> controlFields, AlephBase base) {
        for (int i = 0; i < controlFields.size(); i++) {
            if ("001".equals(controlFields.get(i).getTag())) {
                String sysno = controlFields.get(i).getData();
                return new Identifier(base, sysno);
            }
        }
        return null;
    }

    /**
     * Transforms document from <collection><record> ... </record></collection>
     * to <marc21:record xmlns:marc21="http://www.loc.gov/MARC21/slim"
     * xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     * xsi:schemaLocation="http://www.loc.gov/MARC21/slim
     * http://www.loc.gov/standards/marcxml/schema/MARC21slim.xsd"> ...
     * </marc21:record> Unfortunatelly this solution causes redefinition of
     * default namespece in subelemens of record (for instance <record
     * xmlns="http://www.loc.gov/MARC21/slim"><xmlns=""
     * leader>.....</leader></record> It is caused by long known bug in dom4j:
     * http://www.mail-archive.com/dom4j-dev@lists.sourceforge.net/msg01219.html
     * @param orignal
     * @deprecated
     * @return
     */
    private Document removeCollectionEnvelope(Document original) {
        Element record = (Element) marcRecord.selectSingleNode(original);
        record.detach();
        Document result = DocumentHelper.createDocument();
        result.setRootElement(record);
        record.addNamespace("", MetadataFormat.marc21.getNamespaceUri());
        record.addAttribute(new QName("schemaLocation", xsi), marcSchemaLocation);
        return result;
    }

    /**
     * Transforms document from <collection><record> ... </record></collection>
     * to <marc21:record xmlns:marc21="http://www.loc.gov/MARC21/slim"
     * xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     * xsi:schemaLocation="http://www.loc.gov/MARC21/slim
     * http://www.loc.gov/standards/marcxml/schema/MARC21slim.xsd"> ...
     * </marc21:record>
     *
     * @param orignal
     * @return
     */
    private Document removeCollectionEnvelopeByMakingDocumentCopy(Document original) {
        Namespace marc = new Namespace(MetadataFormat.marc21.name(), MetadataFormat.marc21.getNamespaceUri());
        DocumentFactory factory = DocumentFactory.getInstance();
        Element rootElement = factory.createElement(new QName("record", marc));
        rootElement.addNamespace(xsi.getPrefix(), xsi.getText());
        rootElement.addNamespace(marc.getPrefix(), marc.getText());
        Attribute schemaLocation = factory.createAttribute(rootElement, new QName("schemaLocation", xsi), "http://www.loc.gov/MARC21/slim http://www.loc.gov/standards/marcxml/schema/MARC21slim.xsd");
        rootElement.add(schemaLocation);
        Document doc = factory.createDocument(rootElement);
        Element recordEl = (Element) original.getRootElement().elements().get(0);
        copyElement(rootElement, recordEl, marc, factory);
        return doc;
    }

    /**
     * Copies all attributes, child elements and text nodes from elementFrom to
     * elementTo (both are in different documents). The method also adds
     * namespace prefix.
     *
     * @param elementTo
     * @param elementFrom
     * @param namespace
     * @param factory
     */
    private static void copyElement(Element elementTo, Element elementFrom, Namespace namespace, DocumentFactory factory) {
        //attributes
        Iterator<Attribute> attributes = elementFrom.attributeIterator();
        while (attributes.hasNext()) {
            elementTo.add((Attribute) attributes.next().clone());
        }

        //elements
        Iterator<Element> elements = elementFrom.elementIterator();
        while (elements.hasNext()) {
            Element el = elements.next();
            //System.out.println("adding " + el.getName() + ", " + el.attributeValue("tag"));
            QName qualifiedName = new QName(el.getName(), namespace);
            Element newEl = factory.createElement(qualifiedName);
            elementTo.add(newEl);
            copyElement(newEl, el, namespace, factory);
        }

        //text nodes
        String text = elementFrom.getText();
        //System.out.println("text: " + text);
        if (text != null && text.length() != 0) {
            elementTo.addText(elementFrom.getText());
        }
    }
}
